package com.my.kafka.demo;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 生产者
 */
public class ProducerDemo {
    public static void main(String[] args) {
        //1.kafka的配置信息
        Properties pro = new Properties();

        //Kafka的连接地址
        pro.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"49.234.52.192:9092");
        //发送失败，失败重连次数
        pro.put(ProducerConfig.RETRIES_CONFIG,5);
        //消息key的序列化器
        pro.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //消息value的序列化器
        pro.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //数据压缩
        pro.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");

        //2.生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(pro);

        for (int i = 0; i < 5; i++) {
            //3.封装发送消息
            ProducerRecord<String, String> message = new ProducerRecord<>("tbug-topic-input", "ni gan ma ai yuo");

            //4.发送消息
            producer.send(message);
        }

        //5.关闭消息通道(必选)
        producer.close();
    }
}
